package com.sunzm.flink.datastream.scala.time

import java.time.Duration

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.{DateUtils, FastDateFormat}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, Watermark, WatermarkGenerator, WatermarkGeneratorSupplier, WatermarkOutput, WatermarkStrategy}
import org.apache.flink.streaming.api.TimerService
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 * 多个并行度下面自定义WaterMark示例程序(运行5秒乱序数据)
 */
object ScalaCusWartermarkDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {
    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //设置全局并行度为3
    env.setParallelism(3)

    //设置周期性水印(水位线/Watermark)的间隔,默认200毫秒
    env.getConfig.setAutoWatermarkInterval(500L)

    //从socket端口获取数据
    val dataStream: DataStream[CallLog] = env.socketTextStream("82.156.210.70", 9999)
    //因为socketTextStream并行度只能为1，所以先对数据处理下，后面才能演示
    //或者可以直接使用 kafka 作为数据源也行
   .flatMap((line, out: Collector[CallLog]) => {
      if(StringUtils.isNotBlank(line)){

        try {
          val jSONObject: JSONObject = JSON.parseObject(line)

          val callId: String = jSONObject.getString("callId")
          val dateTimeStr: String = jSONObject.getString("dateTimeStr")

          out.collect(CallLog(callId, dateTimeStr))
        } catch {
          case e: Throwable => {
            logger.warn(s"收到格式异常数据: ${line}", e)
          }
        }
      }
    })

    ascWaterMarkGen(dataStream)

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }

  def ascWaterMarkGen(dataStream: DataStream[CallLog]) = {
    //给数据流设置 时间时间提取器和watermark生成策略
    dataStream.assignTimestampsAndWatermarks(new WatermarkStrategy[CallLog]{
      override def createWatermarkGenerator(context: WatermarkGeneratorSupplier.Context) = new WatermarkGenerator[CallLog]{
        private var maxTimestamp: Long = 0L
        //允许5秒延迟
        private val delay: Long = 5000
        //记录最后一次更新 maxTimestamp 的时间
        private var lastUpdateTimestamp: Long = 0

        override def onEvent(event: CallLog, eventTimestamp: Long, output: WatermarkOutput) = {
          maxTimestamp = Math.max(maxTimestamp, eventTimestamp)
          lastUpdateTimestamp = System.currentTimeMillis()
        }

        override def onPeriodicEmit(output: WatermarkOutput) = {
          if(System.currentTimeMillis() - lastUpdateTimestamp > Duration.ofMinutes(10).toMillis){
            //如果距离上次收到消息已经过去10分钟，就把Watermark推进一下
            maxTimestamp += 1
          }
          output.emitWatermark(new Watermark(maxTimestamp - delay))
        }
      }
    }
      //如果20秒没收到数据，只要任意一个分区Watermark到了触发时间，就会执行
      .withIdleness(Duration.ofSeconds(20)))
      .keyBy(_.callId)
      //开始处理数据
      .process(new KeyedProcessFunction[String, CallLog, String] {

        override def processElement(value: CallLog,
                                    ctx: KeyedProcessFunction[String, CallLog, String]#Context,
                                    out: Collector[String]) = {

          //事件时间
          val eventTime: Long = ctx.timestamp()

          val timerService: TimerService = ctx.timerService()

          //当前的 Watermark
          //刚启动是 Watermark 等于Long的最小值
          val currentWatermark: Long = timerService.currentWatermark()

          //以事件时间为参考点，创建一个5秒后执行的定时器
          val timerMillis: Long = eventTime + Duration.ofSeconds(5).toMillis

          val fastDateFormat: FastDateFormat = FastDateFormat.getInstance("yyyyMMddHHmmssSSS")

          val eventTimeStr: String = fastDateFormat.format(eventTime)
          val currentWatermarkStr: String = if(currentWatermark < 0) "初始值" else fastDateFormat.format(currentWatermark)
          val timerMillisStr: String = fastDateFormat.format(timerMillis)

          """
            |
            |""".stripMargin
          logger.info(
            s""" 收到消息: ${value},
                 |当前事件时间: ${eventTimeStr},
                 |当前Watermark时间: ${currentWatermarkStr},
                 |注册的定时器时间: ${timerMillisStr},
               """.stripMargin)

          timerService.registerEventTimeTimer(timerMillis)

        }

        override def onTimer(timestamp: Long,
                             ctx: KeyedProcessFunction[String, CallLog, String]#OnTimerContext,
                             out: Collector[String]) = {

          val fastDateFormat: FastDateFormat = FastDateFormat.getInstance("yyyyMMddHHmmssSSS")

          val timestampStr: String = fastDateFormat.format(timestamp)

          val currentWatermark: Long = ctx.timerService().currentWatermark()
          val currentWatermarkStr: String = fastDateFormat.format(currentWatermark)

          logger.info(s"定时器触发: ${timestampStr}, 当前Watermark: ${currentWatermarkStr}")

          out.collect(s"${ctx.getCurrentKey} 超时...")

        }
      }).print("outPut")
  }

  //封装数据的样例类
  private case class CallLog(callId: String, dateTimeStr: String){
    override def toString = callId + "," + dateTimeStr
  }
}
